
############################################################################################################
#
#  Date          Author          Changes
#  2024/09/14    veenus.yu       Created
#
###########################################################################################################
import serial_module 
import serial
import time
import os
import sys
import re
import datetime
import fnmatch
import subprocess
import random
import threading


class Stress_script(object):
	"""docstring for fourRouteAPN"""
	def __init__(self,**ports):
     
		self.serial_obj = serial_module.serialObj(**ports)
		self.at_log            = "AT_LOG_{}.log".format(time.strftime("%Y-%m-%d_%H_%M_%S", time.localtime()))
		self.debug_log         = "DEBUG_LOG_{}.log".format(time.strftime("%Y-%m-%d_%H_%M_%S", time.localtime()))
		self.aux_at_log            = "AUX_AT_LOG_{}.log".format(time.strftime("%Y-%m-%d_%H_%M_%S", time.localtime()))
		self.aux_debug_log     = "AUX_DEBUG_LOG_{}.log".format(time.strftime("%Y-%m-%d_%H_%M_%S", time.localtime()))
		self.result_log        = "RESULT_LOG_{}.log".format(time.strftime("%Y-%m-%d_%H_%M_%S", time.localtime()))


	def writeRuntimesToLog_at(self,runTimes):
		with open(self.at_log, 'a') as file_object:
			file_object.write("\r\n{}    runTime:{} {} {}\r\n".format("<<"*5,str(runTimes),time.strftime("%Y-%m-%d_%H_%M_%S", time.localtime()),">>"*5))

	def writeRuntimesToLog_debug(self,runTimes):
		with open(self.debug_log, 'a') as file_object:
			file_object.write("\r\n{}    runTime:{} {} {}\r\n".format("<<"*5,str(runTimes),time.strftime("%Y-%m-%d_%H_%M_%S", time.localtime()),">>"*5))
   
	def writeRuntimesToLog_aux_debug(self,runTimes):
		with open(self.aux_debug_log, 'a') as file_object:
			file_object.write("\r\n{}    runTime:{} {} {}\r\n".format("<<"*5,str(runTimes),time.strftime("%Y-%m-%d_%H_%M_%S", time.localtime()),">>"*5))
   
	def writeRuntimesToLog_aux_at(self,runTimes):
		with open(self.aux_at_log, 'a') as file_object:
			file_object.write("\r\n{}    runTime:{} {} {}\r\n".format("<<"*5,str(runTimes),time.strftime("%Y-%m-%d_%H_%M_%S", time.localtime()),">>"*5))

	def writeRuntimesToLog_result(self,runTimes):
		with open(self.result_log, 'a') as file_object:
			file_object.write("\r\n{}    runTime:{} {} {}\r\n".format("<<"*5,str(runTimes),time.strftime("%Y-%m-%d_%H_%M_%S", time.localtime()),">>"*5))

	def IntoAdbShell(adbID):
		process = subprocess.Popen("adb -s {} shell ".format(adbID), shell=True, stdin=subprocess.PIPE, stdout=subprocess.PIPE,
									stderr=subprocess.PIPE, encoding="utf-8")
		return process

	#检查A模块是否登录
	def login_Device(self):
		curr_time = datetime.datetime.now()
		print("[{}] A模组开始登录...\n".format(time.strftime("%Y-%m-%d_%H_%M_%S", time.localtime())))
		while True:
			time.sleep(1)
			self.serial_obj.serial_write(self.serial_obj.debug_serial,"\r\n")
			result = self.serial_obj.serial_read(self.serial_obj.debug_serial, self.debug_log)
			if "#" in result:
				print('[{}] A模组已经是登录状态\n'.format(time.strftime("%Y-%m-%d_%H_%M_%S", time.localtime())))
				self.serial_obj.serial_write(self.serial_obj.debug_serial,"\003\r\n")
				self.serial_obj.serial_read(self.serial_obj.debug_serial, self.debug_log)
				break

			if "login:" in result:
				self.serial_obj.serial_write(self.serial_obj.debug_serial,"root\r\n")
				time.sleep(1)
				self.serial_obj.serial_read(self.serial_obj.debug_serial, self.debug_log)
				# self.serial_obj.serial_write(self.serial_obj.debug_serial,"fmOlFugB5vYQqM2ugwE-EJAFaBkPiU0DBLxwW1adToai8qKA5NXYKAMotNHNn5XMjH4pzieLzh9e-JsLfTEo1zgBozMiSc-x3CnO8luM0aSNFun1NgbgBOKkDf-S1tfC\r\n")
				time.sleep(1)
				self.serial_obj.serial_read(self.serial_obj.debug_serial, self.debug_log)
    
			if "~ #" in result:
				print('[{}] A模块登录成功\n'.format(time.strftime("%Y-%m-%d_%H_%M_%S", time.localtime())))
				break
			else:
				self.serial_obj.serial_write(self.serial_obj.debug_serial,"\003\r\n")

			tmp_time = datetime.datetime.now()
			if (tmp_time - curr_time).seconds >200:
				print("[{}] A模块登录失败\n".format(time.strftime("%Y-%m-%d_%H_%M_%S", time.localtime())))
				sys.exit()

	
	# 检查B模块是否登录
	def aux_login_Device(self):
		curr_time = datetime.datetime.now()
		print("[{}] B模组开始登录...\n".format(time.strftime("%Y-%m-%d_%H_%M_%S", time.localtime())))
		while True:
			time.sleep(1)
			self.serial_obj.serial_write(self.serial_obj.aux_debug_port,"\r\n")
			result = self.serial_obj.serial_read(self.serial_obj.aux_debug_port, self.aux_debug_log)
			if "#" in result:
				print('[{}] B模组已经是登录状态\n'.format(time.strftime("%Y-%m-%d_%H_%M_%S", time.localtime())))
				self.serial_obj.serial_write(self.serial_obj.aux_debug_port,"\003\r\n")
				self.serial_obj.serial_read(self.serial_obj.aux_debug_port, self.aux_debug_log)
				break

			if "login:" in result:
				self.serial_obj.serial_write(self.serial_obj.aux_debug_port,"root\r\n")
				time.sleep(1)
				self.serial_obj.serial_read(self.serial_obj.aux_debug_port, self.aux_debug_log)
				# self.serial_obj.serial_write(self.serial_obj.aux_debug_port,"fmOlFugB5vYQqM2ugwE-EJAFaBkPiU0DBLxwW1adToai8qKA5NXYKAMotNHNn5XMjH4pzieLzh9e-JsLfTEo1zgBozMiSc-x3CnO8luM0aSNFun1NgbgBOKkDf-S1tfC\r\n")
				time.sleep(1)
				self.serial_obj.serial_read(self.serial_obj.aux_debug_port, self.aux_debug_log)
    
			if "~ #" in result:
				print('[{}] B模组登录成功\n'.format(time.strftime("%Y-%m-%d_%H_%M_%S", time.localtime())))
				break
			else:
				self.serial_obj.serial_write(self.serial_obj.aux_debug_port,"\003\r\n")

			tmp_time = datetime.datetime.now()
			if (tmp_time - curr_time).seconds >200:
				print("[{}] B模块登录失败\n".format(time.strftime("%Y-%m-%d_%H_%M_%S", time.localtime())))
				sys.exit()
	def checkNetWork_A(self):
		i = 0
		while i<30:
			self.serial_obj.serial_write(self.serial_obj.at_serial,"AT+CREG?;+CGREG?;+CEREG?;+C5GREG?;+QCSQ;+QNWINFO;+COPS?;\r\n")
			time.sleep(2)
			response_reg = self.serial_obj.serial_read(self.serial_obj.at_serial,self.at_log)
			# if ("+CEREG: 0,1" in response_reg and "NR5G-SA"  in response_reg) or ("+CEREG: 0,1" in response_reg and "LTE"  in response_reg):
			if "NR5G-SA"  in response_reg:
				print("[{}] A模块注网正常.\n".format(time.strftime("%Y-%m-%d_%H_%M_%S", time.localtime())))
				break
			i += 1
		if i==30:
			print("[{}] A模块注网不正常，failed.\n".format(time.strftime("%Y-%m-%d_%H_%M_%S", time.localtime())))
			sys.exit()	
        
	def checkNetWork_B(self):
		i = 0
		while i<30:
			self.serial_obj.serial_write(self.serial_obj.aux_at_serial,"AT+CREG?;+CGREG?;+CEREG?;+C5GREG?;+QCSQ;+QNWINFO;+COPS?;\r\n")
			time.sleep(2)
			response_reg = self.serial_obj.serial_read(self.serial_obj.aux_at_serial,self.aux_at_log)
			if ("+CEREG: 0,1" in response_reg and "NR5G-SA"  in response_reg) or ("+CEREG: 0,1" in response_reg and "LTE"  in response_reg):
				print("[{}] B模块注网正常.\n".format(time.strftime("%Y-%m-%d_%H_%M_%S", time.localtime())))
				break
			i += 1
		if i==30:
			print("[{}] B模块注网不正常，failed.\n".format(time.strftime("%Y-%m-%d_%H_%M_%S", time.localtime())))
			sys.exit()
    #A模块退出telsdk_console_app
	def exit_telsdk_console_app_A(self):
		self.serial_obj.serial_write(self.serial_obj.debug_serial,"exit\r")
		time.sleep(1)
		self.serial_obj.serial_write(self.serial_obj.debug_serial,"exit\r")
		time.sleep(1)
		exit_result = self.serial_obj.serial_read(self.serial_obj.debug_serial, self.debug_log)
		if "#" in exit_result:
			print("[{}] 退出telsdk_console_app成功\n".format(time.strftime("%Y-%m-%d_%H_%M_%S", time.localtime())))
		else:
			print("[{}] 退出telsdk_console_app失败\n".format(time.strftime("%Y-%m-%d_%H_%M_%S", time.localtime())))
   #B模块退出telsdk_console_app
	def exit_telsdk_console_app_B(self):
		self.serial_obj.serial_write(self.serial_obj.aux_debug_port,"exit\r")
		time.sleep(1)
		self.serial_obj.serial_write(self.serial_obj.aux_debug_port,"exit\r")
		time.sleep(1)
		exit_result = self.serial_obj.serial_read(self.serial_obj.aux_debug_port, self.aux_debug_log)
		if "#" in exit_result:
			print("[{}] 退出telsdk_console_app成功\n".format(time.strftime("%Y-%m-%d_%H_%M_%S", time.localtime())))
		else:
			print("[{}] 退出telsdk_console_app失败\n".format(time.strftime("%Y-%m-%d_%H_%M_%S", time.localtime())))
		
    #模块A发送短信		
	def create_send_msg_A_B(self):
		# 运行telsdk_console_app	
		self.serial_obj.serial_write(self.serial_obj.debug_serial,"telsdk_console_app\r\n")
		time.sleep(1)
		sdk_result = self.serial_obj.serial_read(self.serial_obj.debug_serial, self.debug_log)
		if "SMS" in sdk_result:
			print("[{}] 运行telsdk_console_app成功\n".format(time.strftime("%Y-%m-%d_%H_%M_%S", time.localtime())))
		else:
			print("[{}] 运行telsdk_console_app失败\n".format(time.strftime("%Y-%m-%d_%H_%M_%S", time.localtime())))
			sys.exit()
		# 进入sms接口
		self.serial_obj.serial_write(self.serial_obj.debug_serial,"4\r")
		time.sleep(2)
		
		self.serial_obj.serial_write(self.serial_obj.debug_serial,"1\r")
		time.sleep(2)
		self.serial_obj.serial_write(self.serial_obj.debug_serial,"{}\r".format(phoneNumB))
		time.sleep(2)
		self.serial_obj.serial_write(self.serial_obj.debug_serial,"1212121212\r")
		time.sleep(2)
		self.serial_obj.serial_write(self.serial_obj.debug_serial,"y\r")
		sms_result = self.serial_obj.serial_read(self.serial_obj.debug_serial, self.debug_log)
		if "Send SMS request successful" in sms_result:
			print("[{}] 模块A发送短信成功\n".format(time.strftime("%Y-%m-%d_%H_%M_%S", time.localtime())))
			curr_time = datetime.datetime.now()
			return curr_time
		else:
			print("[{}] 模块A发送短信失败\n".format(time.strftime("%Y-%m-%d_%H_%M_%S", time.localtime())))
		# print(sms_result)

		
	# 模块A打电话
	def create_call_A_B(self):
		# 运行telsdk_console_app
		self.serial_obj.serial_write(self.serial_obj.debug_serial,"telsdk_console_app\r\n")
		time.sleep(1)
		sdk_result = self.serial_obj.serial_read(self.serial_obj.debug_serial, self.debug_log)
		if "Dialer" in sdk_result:
			print("[{}] 运行telsdk_console_app成功\n".format(time.strftime("%Y-%m-%d_%H_%M_%S", time.localtime())))
		else:
			print("[{}] 运行telsdk_console_app失败\n".format(time.strftime("%Y-%m-%d_%H_%M_%S", time.localtime())))
			sys.exit()
        # 进入Dialer接口
		self.serial_obj.serial_write(self.serial_obj.debug_serial,"2\r")
		time.sleep(2)
		self.serial_obj.serial_write(self.serial_obj.debug_serial,"1 {}\r".format(phoneNumB))
		time.sleep(2)
		Dialer_result = self.serial_obj.serial_read(self.serial_obj.debug_serial, self.debug_log)
		if "Outgoing call" in Dialer_result:
			print("[{}] 模块A拨打电话成功\n".format(time.strftime("%Y-%m-%d_%H_%M_%S", time.localtime())))
			curr_time = datetime.datetime.now()
			return curr_time
		else:
			print("[{}] 模块A拨打电话失败\n".format(time.strftime("%Y-%m-%d_%H_%M_%S", time.localtime())))
		time.sleep(2)
  
  
  
	# 模块A停止打电话
	def create_stop_call_A_B(self):
		alert_result = self.serial_obj.serial_read(self.serial_obj.debug_serial, self.debug_log)
		# print(alert_result)
		if "Alerting call" in alert_result:
			print("[{}] 电话响铃\n".format(time.strftime("%Y-%m-%d_%H_%M_%S", time.localtime())))
        # 响铃等待10S挂断
		print("[{}]响铃等待10S挂断\n".format(time.strftime("%Y-%m-%d_%H_%M_%S", time.localtime())))
		time.sleep(10)
		self.serial_obj.serial_write(self.serial_obj.debug_serial,"5\r")
		

	#B模块进入休眠模式
	def create_sleep_B(self):
        # 打开内核日志：cat /proc/kmsg &
		self.serial_obj.serial_write(self.serial_obj.aux_debug_port,"cat /proc/kmsg &\r\n")
		time.sleep(1)
		# 执行echo mem > /sys/power/autosleep，使能autosleep
		self.serial_obj.serial_write(self.serial_obj.aux_debug_port,"echo mem > /sys/power/autosleep\r\n")
		time.sleep(1)
		# 执行 telux_power_test_app -s 进入休眠模式，拔掉USB，模块进入休眠
		self.serial_obj.serial_write(self.serial_obj.aux_debug_port,"telux_power_test_app -s\r\n")
		time.sleep(1)
		print("[{}]请把usb拔掉\n".format(time.strftime("%Y-%m-%d_%H_%M_%S", time.localtime())))
		# 等待模块进入休眠模式
		time.sleep(10)
		sleep_result = self.serial_obj.serial_read(self.serial_obj.aux_debug_port, self.aux_debug_log)
		if "deep" in sleep_result:
			print("[{}] 模块B进入休眠模式成功\n".format(time.strftime("%Y-%m-%d_%H_%M_%S", time.localtime())))
		else:	
			print("[{}] 模块B进入休眠模式失败\n".format(time.strftime("%Y-%m-%d_%H_%M_%S", time.localtime())))
		# print(sleep_result)
    #判断B 模块是否退出休眠模式
	def check_sleep_B(self):
		flag = 0
		while True:
			flag += 1
			sleep_result = self.serial_obj.serial_read(self.serial_obj.aux_debug_port, self.aux_debug_log)
			self.serial_obj.serial_write(self.serial_obj.aux_debug_port,"\r")
			if "#" in sleep_result:
				print("[{}] 模块B退出休眠模式\n".format(time.strftime("%Y-%m-%d_%H_%M_%S", time.localtime())))
                #记录当前时间
				curr_time = datetime.datetime.now()
				return curr_time
			else:
				print("[{}] 模块未退出休眠模式\n".format(time.strftime("%Y-%m-%d_%H_%M_%S", time.localtime())))
			if flag == 30:
				break
			time.sleep(1)
def get_valid_time_diff(time1, time2, decimal_places=4):
    time_diff = time2 - time1
    total_seconds = time_diff.total_seconds()
    valid_time_diff = round(total_seconds % 1, decimal_places)
    print("[{}]总耗时:{}".format(time.strftime("%Y-%m-%d_%H_%M_%S", time.localtime()),valid_time_diff))
    return valid_time_diff
#============================================================================================#
# 端口定义
# A模块端口
AT_PORT       = "COM39"
DEBUG_PORT    = "COM4"

# B模块端口 
AUX_AT_PORT =  "COM17"
AUX_DEBUG_PORT    = "COM47"

# A模块号码
phoneNumA = "13053133078"

# B模块号码
phoneNumB = "18256942153"

#记录运行次数
runTimes   = 0
module = Stress_script(debug_port= DEBUG_PORT, at_port = AT_PORT, aux_debug_port= AUX_DEBUG_PORT, aux_at_port = AUX_AT_PORT)
# module = Stress_script(debug_port= DEBUG_PORT, at_port = AT_PORT)
time.sleep(2)
module.login_Device()
module.aux_login_Device()
module.checkNetWork_A()
module.checkNetWork_B()
module.create_sleep_B()
while True:
	time_msg = module.create_send_msg_A_B()
	time1 = module.check_sleep_B()
	get_valid_time_diff(time_msg, time1)
	module.exit_telsdk_console_app_A()
	time_call = module.create_call_A_B()
	time2 = module.check_sleep_B()
	get_valid_time_diff(time_call, time2)
	module.create_stop_call_A_B()
	module.exit_telsdk_console_app_A()
	time.sleep(10*60)


